package abacab.remote.oio;

import java.io.DataInputStream;
import java.io.IOException;
import java.net.Socket;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import abacab.remote.Connection;

/**
 * The thread in charge of reading from a socket
 */
public final class SocketReader implements Runnable {
	/** The logger */
	private static final Logger		logger	= LoggerFactory.getLogger(SocketReader.class);

	/** The connection */
	private final Connection			connection;

	/** The input stream */
	private final DataInputStream	inputStream;

	/** Indicates if the reader is stopped */
	private volatile boolean			stopped;

	/**
	 * Create the reader
	 * 
	 * @param connection
	 *          the connection
	 * @param socket
	 *          the socket
	 * @throws IOException
	 *           if an error occurred
	 */
	public SocketReader(Connection connection, Socket socket) throws IOException {
		this.connection = connection;
		this.inputStream = new DataInputStream(socket.getInputStream());
		Thread t = new Thread(this, "socket reader: " + socket.getRemoteSocketAddress());
		t.setDaemon(true);
		t.start();
	}

	/**
	 * Stop the reader
	 */
	public void stop() {
		this.stopped = true;
	}

	@Override
	public void run() {
		byte[] sizeBuffer = new byte[2];
		try {
			while (!this.stopped) {
				this.inputStream.readFully(sizeBuffer);
				int size = ((sizeBuffer[0] & 0xFF) << 8) + (sizeBuffer[1] & 0xFF);
				if (size < 0) {
					throw new IOException("Stream corrupted: received a negative data size");
				}
				if (size > 10 * 1024 * 1024) {
					throw new IOException("Stream corrupted: received a data size greater than 10MB (" + size + ")");
				}
				byte[] dataBuffer = new byte[size];
				this.inputStream.readFully(dataBuffer);
				this.connection.messageReceived(dataBuffer);
			}
		}
		catch (IOException e) {
			if (!this.stopped) {
				logger.error("Failed to read from the remote system", e);
				this.connection.close();
			}
		}
	}
}